package com.huan.network;


import com.huan.bean.ApacheLogEvent;
import com.huan.bean.PageViewCount;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.regex.Pattern;

//乱序数据测试
public class HotPages_Watermarks {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );
        env.setParallelism( 1 );

        DataStream<String> inputStream = env.socketTextStream( "localhost", 7777 );


        //格式化输出
        DataStream<ApacheLogEvent> dataStream = inputStream
                .map( line -> {
                    String[] fields = line.split( " " );
                    //定义时间，转化时间格式
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat( "dd/MM/yyyy:HH:mm:ss" );
                    //解析时间并且获取
                    Long timestamp = simpleDateFormat.parse( fields[3] ).getTime();
                    return new ApacheLogEvent( fields[0], fields[1], timestamp, fields[5], fields[6] );
                } )
                .assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor<ApacheLogEvent>( Time.seconds( 1 ) ) {
                    @Override
                    public long extractTimestamp(ApacheLogEvent element) {
                        return element.getTimestamp();
                    }
                } );

        //完整数据
        dataStream.print("data");

        /**
         * 数据每隔十分钟的一个窗口，隔五秒滑动一次，统计打印滑动出来的结果，因为有Watermarks机制，窗口延迟一秒钟触发操作
         * 有了Watermarks延迟之后，再过一秒则展示界面，
         * [50,55) 遵守左闭右开原则，当前面输入 50，51，52，53，54秒的数据 ，则是[50,55)之间的数据，当输入55的时候
         * 将[50.55)之间的数据做一个汇总，当输出56时候，将[50,55)之间的数据格式化打印到控制台
         * 并且允许迟到数据一分钟
         */

        // 定义一个侧输出流标签
        OutputTag<ApacheLogEvent> lateTag = new OutputTag<ApacheLogEvent>("late"){};

        //分组开窗聚合
        SingleOutputStreamOperator<PageViewCount> windowAggStream = dataStream
                .filter(data -> "GET".equals(data.getMethod()))    // 过滤get请求
                .filter(data -> {
                    String regex = "^((?!\\.(css|js|png|ico)$).)*$";
                    return Pattern.matches(regex, data.getUrl());
                })
                .keyBy(ApacheLogEvent::getUrl)    // 按照url分组
                .timeWindow(Time.minutes(10), Time.seconds(5))
                //设置允许元素延迟的时间。到达水印后超过指定时间的元素将被删除。
                .allowedLateness(Time.minutes(1))
                //侧输出流
                .sideOutputLateData(lateTag)
                .aggregate(new PageCountAgg(), new PageCountResult());


        windowAggStream.print("agg");
        windowAggStream.getSideOutput(lateTag).print("late");

        // 收集同一窗口count数据，排序输出
        DataStream<String> resultStream = windowAggStream
                .keyBy(PageViewCount::getWindowEnd)
                .process(new TopNHotPages(3));

        resultStream.print();

        env.execute("HotPages_Watermarks");
    }

}
